package com.hmdp.service.impl;

import cn.hutool.core.bean.BeanUtil;
import com.hmdp.dto.Result;
import com.hmdp.entity.VoucherOrder;
import com.hmdp.mapper.VoucherOrderMapper;
import com.hmdp.service.ISeckillVoucherService;
import com.hmdp.service.IVoucherOrderService;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.hmdp.utils.RedisIdWorker;
import com.hmdp.utils.UserHolder;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RedissonClient;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.aop.framework.AopContext;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.io.ClassPathResource;
import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import javax.annotation.Resource;
import java.time.Duration;
import java.time.LocalDateTime;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;

import static com.hmdp.utils.MqConstance.*;


@Service
@Slf4j
public class VoucherOrderServiceImpl extends ServiceImpl<VoucherOrderMapper, VoucherOrder> implements IVoucherOrderService {

    @Resource
    private ISeckillVoucherService seckillVoucherService;
    @Autowired
    private RedisIdWorker redisIdWorker;
    @Autowired
    private StringRedisTemplate stringRedisTemplate;
    @Autowired
    private RabbitTemplate rabbitTemplate;
    /*@Resource
    private RedissonClient redissonClient;*/

    private IVoucherOrderService proxy;
    //阻塞队列
    //private BlockingQueue<VoucherOrder> orderTasks = new ArrayBlockingQueue<>(1024 * 1024);

    private static final DefaultRedisScript<Long> SECKILL_SCRIPT;

    static {
        //初始化lua脚本
        SECKILL_SCRIPT = new DefaultRedisScript<>();
        SECKILL_SCRIPT.setLocation(new ClassPathResource("seckill.lua"));
        SECKILL_SCRIPT.setResultType(Long.class);

    }

   /* //异步处理线程池
    private static final ExecutorService SECKILL_ORDER_EXECUTOR = Executors.newSingleThreadExecutor();

    @PostConstruct //在类初始化之后执行，因为当这个类初始化好了之后，随时都会有订单产生
    private void init() {
        SECKILL_ORDER_EXECUTOR.submit(new VoucherOrderHandler());
    }*/

    @Override
    public Result seckillVoucher(Long voucherId) {
        //获取用户
        Long userId = UserHolder.getUser().getId();
        long orderId = redisIdWorker.nextId("order");
        // 1.执行lua脚本
        Long result = stringRedisTemplate.execute(
                SECKILL_SCRIPT,
                Collections.emptyList(),
                voucherId.toString(), userId.toString()
        );
        int r = result.intValue();
        // 2.判断结果是否为0
        if (r != 0) {
            // 2.1.不为0 ，代表没有购买资格
            log.error(r == 1 ? "库存不足" : "不能重复下单");
            return Result.fail(r == 1 ? "库存不足" : "不能重复下单");
        }

        //有购买资格创建订单
        VoucherOrder voucherOrder = new VoucherOrder();
        voucherOrder.setId(orderId);
        voucherOrder.setUserId(userId);
        voucherOrder.setVoucherId(voucherId);
        createVoucherOrder(voucherOrder);

        //3.获取代理对象
        //proxy = (IVoucherOrderService) AopContext.currentProxy();//只有代理对象才能走事务并且只有主线程才能获取到代理对象

        //4.返回订单id
        return Result.ok(orderId);
    }


    // 异步线程任务当初始化完毕后，就会去从对列中去拿信息
  /*  private class VoucherOrderHandler1 implements Runnable {
        @Override
        public void run() {
            while (true) {
            try {
                // 1.获取消息队列中的订单信息 XREADGROUP GROUP g1 c1 COUNT 1 BLOCK 2000 STREAMS s1 > Consumer.from如果没有消费者组和消费者则会创建
                    List<MapRecord<String, Object, Object>> list = stringRedisTemplate.opsForStream().read(Consumer.from("g1", "c1"),
                           StreamReadOptions.empty().count(1).block(Duration.ofSeconds(2)),//empty()的作用是创建一个空的StreamReadOptions对象然后在这个对象上调用count()方法
                            StreamOffset.create("stream.orders", ReadOffset.lastConsumed()));

                    // 2.判断订单信息是否为空
                    if(list == null || list.isEmpty()){
                        continue;
                    }

                MapRecord<String, Object, Object> record = list.get(0);
                Map<Object, Object> value = record.getValue();
                VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(value, new VoucherOrder(), true);
                // 3.创建订单
                proxy.createVoucherOrder(voucherOrder);
                log.info("订单创建成功：{}", voucherOrder);
                // 4.确认消息 XACK acknowledge的第三个参数的作用是确认信息具体是哪个信息
                stringRedisTemplate.opsForStream().acknowledge("s1", "g1", record.getId());


            } catch (Exception e) {
                log.error("处理订单异常", e);
                //处理异常消息 在上述过程中出现异常最终导致消息没有被消费重新创建订单
                handlePendingList1();
            }
            }
        }
    }*/
  /*  private void handlePendingList1() {
        while (true) {
        try {
            // 1.获取pending-list中的订单信息 XREADGROUP GROUP g1 c1 COUNT 1 BLOCK 2000 STREAMS s1 0
            List<MapRecord<String, Object, Object>> list = stringRedisTemplate.opsForStream().read(
                    Consumer.from("g1", "c1"),
                    StreamReadOptions.empty().count(1),
                    StreamOffset.create("stream.orders", ReadOffset.from("0")));

            //2.判断订单信息是否为空 理论上是肯定有异常的订单才会进来
                if(list == null || list.isEmpty()){
                    break;
                }

            MapRecord<String, Object, Object> record = list.get(0);
            Map<Object, Object> value = record.getValue();
            VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(value, new VoucherOrder(), true);
            // 3.创建订单
            proxy.createVoucherOrder(voucherOrder);
            // 4.确认消息 XACK
            stringRedisTemplate.opsForStream().acknowledge("s1", "g1", record.getId());
        } catch (Exception e) {
                log.error("处理pendding订单异常", e);
                try {
                    Thread.sleep(20);
                } catch (Exception ex) {
                    e.printStackTrace();
                }
            }
        }

    }*/
    private class VoucherOrderHandler implements Runnable {

        @Override
        public void run() {
            while (true) {
                try {
                    // 1.获取消息队列中的订单信息 XREADGROUP GROUP g1 c1 COUNT 1 BLOCK 2000 STREAMS s1 >
                    @SuppressWarnings("unchecked")
                    List<MapRecord<String, Object, Object>> list = stringRedisTemplate.opsForStream().read(
                            Consumer.from("g1", "c1"),
                            StreamReadOptions.empty().count(1).block(Duration.ofSeconds(2)),
                            StreamOffset.create("stream.orders", ReadOffset.lastConsumed())
                    );
                    // 2.判断订单信息是否为空
                    if (list == null || list.isEmpty()) {
                        // 如果为null，说明没有消息，继续下一次循环
                        continue;
                    }
                    // 解析数据
                    MapRecord<String, Object, Object> record = list.get(0);
                    Map<Object, Object> value = record.getValue();
                    VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(value, new VoucherOrder(), true);
                    // 3.创建订单
                    proxy.createVoucherOrder(voucherOrder);
                    // 4.确认消息 XACK
                    stringRedisTemplate.opsForStream().acknowledge("s1", "g1", record.getId());
                } catch (Exception e) {
                    log.error("处理订单异常", e);
                    //处理异常消息
                    handlePendingList();
                }
            }
        }

        private void handlePendingList() {
            while (true) {
                try {
                    // 1.获取pending-list中的订单信息 XREADGROUP GROUP g1 c1 COUNT 1 BLOCK 2000 STREAMS s1 0
                    List<MapRecord<String, Object, Object>> list = stringRedisTemplate.opsForStream().read(
                            Consumer.from("g1", "c1"),
                            StreamReadOptions.empty().count(1),
                            StreamOffset.create("stream.orders", ReadOffset.from("0"))
                    );
                    // 2.判断订单信息是否为空
                    if (list == null || list.isEmpty()) {
                        // 如果为null，说明没有异常消息，结束循环
                        break;
                    }
                    // 解析数据
                    MapRecord<String, Object, Object> record = list.get(0);
                    Map<Object, Object> value = record.getValue();
                    VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(value, new VoucherOrder(), true);
                    // 3.创建订单
                    proxy.createVoucherOrder(voucherOrder);
                    // 4.确认消息 XACK
                    stringRedisTemplate.opsForStream().acknowledge("s1", "g1", record.getId());
                } catch (Exception e) {
                    log.error("处理pendding订单异常", e);
                    try{
                        Thread.sleep(1000);
                    }catch(Exception ex){
                        ex.printStackTrace();
                    }
                }
            }
        }
    }


    //private void handleVoucherOrder(VoucherOrder voucherOrder) {
    //1.获取用户注意因为是异步线程，不是主线程获取不到用户信息
//        Long userId = voucherOrder.getUserId();
//        // 2.创建锁对象
//        RLock redisLock = redissonClient.getLock("lock:order:" + userId);
//        // 3.尝试获取锁
//        boolean isLock = redisLock.tryLock();
//        // 4.判断是否获得锁成功
//        if (!isLock) {
//            // 获取锁失败，直接返回失败或者重试
//            log.error("不允许重复下单！");
//            return;
//        }
//        try {
//            //注意：由于是spring的事务是放在threadLocal中，此时的是多线程，事务会失效
//            proxy.createVoucherOrder(voucherOrder);
//        } finally {
//            // 释放锁
//            redisLock.unlock();
//        }
//    }


    /*
    @Override
    public Result seckillVoucher(Long voucherId) {

        //1.查询优惠券信息
        SeckillVoucher seckillVoucher = seckillVoucherService.getById(voucherId);
        LocalDateTime beginTime = seckillVoucher.getBeginTime();

        //2.判断秒杀是否开始
        if(beginTime.isAfter(LocalDateTime.now())){
            return Result.fail("秒杀尚未开始！");
        }
        LocalDateTime endTime = seckillVoucher.getEndTime();
        if(endTime.isBefore(LocalDateTime.now())){
            return Result.fail("秒杀已经结束！");
        }

        Integer stock = seckillVoucher.getStock();
        //@@@这里有线程安全问题，因为同时有多个请求进来，都会查询到库存大于0，然后都会减库存，导致超卖
        //解决办法 1.乐观锁不加锁并且会有一个版本号在本项目stock可以当作版本号只是在扣减库存之前会进行判断如果改变了就不扣减但是会使失败率增加
        //where id = ？ and stock = 原数值 2.因为在本项目有库存修改后其他线程均不能扣减库存 所以说只需要stock>0就可以避免//where id = ? and stock > 0
        if(stock < 1){
            return Result.fail("库存不足！");
        }
        Long userId = UserHolder.getUser().getId();
        //1.redis创建锁对象
        //ILockImpl lock = new ILockImpl("order:" + userId, stringRedisTemplate);
        //获取锁
        //boolean success = lock.tryLock(1200);
        //使用分布式锁
        //2.redission
        // 具体来说，getLock 方法会根据传入的锁名称生成一个 RLock 对象，然后在调用 tryLock 方法时，
        // RLock 对象会使用这个名称在 Redis 中尝试设置一个键值对来表示锁
        RLock lock = redissonClient.getLock("lock:order:" + userId);
        //获取锁对象
        //参数分别是：获取锁的最大等待时间(期间会重试)，锁自动释放时间，时间单位
        boolean isLock = lock.tryLock();//会尝试在 Redis 中设置一个键值对
        //KEYS[1] = "lock:order:" + userId; ARGV[1]：锁失效时间  ARGV[2]：threadId(sdasdasdkas-33); 锁的小key
        //通过大key+小key判断当前这把锁是否是属于自己的，如果是自己的，则进行重入加一
        //仍然实现一人一单每个用户的线程keyKEYS[1]锁名称都是不同的相同线程的key才是相同的
        //参数分别是：获取锁的最大等待时间(期间会重试)，锁自动释放时间，时间单位
        //根据业务这里无参默认等待时间为-1失败不等待因为我们是实现一人一单，锁自动释放时间默认为30秒才能开启WatchDog
        if(!isLock){
            return Result.fail("不允许重复下单！");
        }
        //1.synchronized锁住的是当前对象，所以在多个对象中是不会有影响的，所以需要锁住的是用户id
//        synchronized (userId.toString().intern()){
//            return this.creatVourcherOrder(voucherId);
//        }

        //2.问题如果想要事务生效，需要使用代理对象调用，因为代理对象调用会走spring的事务控制
        //并且再此之前需要在启动类上加@EnableAspectJAutoProxy(exposeProxy = true)暴露代理
        // 并且需要引入org.aspectj:aspectjweaver依赖
        try{
            IVoucherOrderService proxy = (IVoucherOrderService) AopContext.currentProxy();
            return proxy.creatVourcherOrder(voucherId);
        }
        finally {
            lock.unlock();
        }

    }
     */
//@@@但是目前业务逻辑是需要一人一单需要增加判断
    //增加一人一单逻辑 这里并发同时都是0 即可去扣减所以需要使用悲观锁，1.但是不能在方法上添加了synchronized锁，锁的粒度太粗了
    // 2.锁用户id因为使用类作为id会使所有用户线性执行，而用户id一个用户就一把每个用户都有一把锁，锁针对的是同一用户多买锁住，不影响其他用户
    //因为就算每次是同一用户都是全新的id对象（Long包装类），所以调用toString保证值一样，但是都会生成全新的字符串对象所以字符串常量池中拿取值相同的内存地址返回
    // 这样就可以唯一的用户锁，防止一个用户多把锁
    //3.如果在creatVourcherOrder方法内部加锁当前方法被spring的事务控制，如果你在方法内部加锁，函数执行完以后再由spring去做提交，
    // 而且是两次操作数据库统一提交，一次是减库存，一次是生成订单
    //所以此时其他线程可以进入并且查询订单仍然不存在，所以需要将整个方法加锁，保证事务不会出现问题
    // 可能会导致当前方法事务还没有提交，但是锁已经释放也会导致问题，所以我们选择将当前方法整体包裹起来，确保事务不会出现问题
    //4.集群环境下的并发问题，synchronized落幕只于单机，因为锁是在本地jvm上的每个jvm都有自己的锁监视器，所以同一个用户在不同的jvm第一次请求均可以成功
    // 所以需要使用分布式锁，redis分布式锁，使得每个jvm都可以访问到唯一的redis上的锁，保证只有一个用户可以进入
    @Transactional
    public void createVoucherOrder(VoucherOrder voucherOrder) {//理论上已经不需要加锁了，但是为了兜底还是可以加锁
        Long userId = voucherOrder.getUserId();
        //synchronized(userId.toString().intern())
        //在Lua脚本中已经判断了能否下单和库存是否足够，但在Java代码中再次进行判断是为了双重保障。这样做的原因包括：
        //数据一致性：Lua脚本在Redis中进行判断和操作，但Redis和数据库是两个独立的系统，可能存在数据同步问题。通过在数据库中再次查询，可以确保数据的一致性。
        //防止并发问题：在高并发场景下，可能会有多个请求同时通过Lua脚本的判断。通过在数据库中再次查询，可以进一步确保同一用户不会重复下单。
        //容错机制：如果Redis中的数据出现问题（例如数据丢失或被篡改），通过数据库查询可以作为一种容错机制，确保业务逻辑的正确性。
        //这种双重检查机制可以提高系统的可靠性和数据的准确性。
        Integer count = query().eq("user_id", userId).eq("voucher_id", voucherOrder.getVoucherId()).count();
        if (count > 0) {
            log.error("不能重复下单！");
        }
        //3.减库存 乐观锁比较适合更新数据
        boolean success = seckillVoucherService.update().setSql("stock = stock - 1")
                .eq("voucher_id", voucherOrder.getVoucherId()).gt("stock", 0).update();

        if (!success) {
            log.error("库存不足！");
        }
        rabbitTemplate.convertAndSend(ORDER_EXCHANGE, ORDER_ROUTING_KEY, voucherOrder);


    }
}

    //加乐观锁锁的方法
//    private void createVoucherOrder1(VoucherOrder voucherOrder) {
//        Long userId = voucherOrder.getUserId();
//        Long voucherId = voucherOrder.getVoucherId();
//        // 创建锁对象
//        RLock redisLock = redissonClient.getLock("lock:order:" + userId);
//        // 尝试获取锁
//        boolean isLock = redisLock.tryLock();
//        // 判断
//        if (!isLock) {
//            // 获取锁失败，直接返回失败或者重试
//            log.error("不允许重复下单！");
//            return;
//        }
//
//        try {
//            // 5.1.查询订单
//            int count = query().eq("user_id", userId).eq("voucher_id", voucherId).count();
//            // 5.2.判断是否存在
//            if (count > 0) {
//                // 用户已经购买过了
//                log.error("不允许重复下单！");
//                return;
//            }
//
//            // 6.扣减库存
//            boolean success = seckillVoucherService.update()
//                    .setSql("stock = stock - 1") // set stock = stock - 1
//                    .eq("voucher_id", voucherId).gt("stock", 0) // where id = ? and stock > 0
//                    .update();
//            if (!success) {
//                // 扣减失败
//                log.error("库存不足！");
//                return;
//            }
//
//            // 7.创建订单
//            save(voucherOrder);
//        } finally {
//            // 释放锁
//            redisLock.unlock();
//        }
//    }
